package com.doitedu.core

import com.doitedu.utils.{MyPartitioner, SparkUtil}
import org.apache.spark.HashPartitioner
import org.apache.spark.rdd.RDD

/**
 * @Date: 22.7.1 
 * @Author: HANGGE
 * @qq: 598196583
 * @Tips: 学大数据 ,到多易教育
 * @Description:
 */
object C16_ReduceByKey {
  def main(args: Array[String]): Unit = {

    val sc = SparkUtil.getSparkContext("reduceByKey")
    val dataRDD = sc.textFile("data/word.txt")
    //coalesce 方法只能减少分布  不能增加分区 因为参数2 默认是false 不允许分区
    //  将参数2修改成true
    val wordsRDD = dataRDD.flatMap(_.split("\\s+")).coalesce(3,true)
   /* wordsRDD
      .groupBy(e=>e)
      .map(tp=>(tp._1 , tp._2.size))
      .foreach(println)*/

 /*   wordsRDD.map((_,1))  // KV RDD
      .groupByKey()
      .map(tp=>(tp._1,tp._2.size))
      .foreach(println)*/

 /*   wordsRDD.map((_,1))
      /**
       * x1 第一个值  后续的代表 上次计算的结果
       * x2 第二个值  第三个  第四个...
       */
    //  .reduceByKey((x1,x2)=>x1+x2)
      .reduceByKey(_+_)
      .foreach(println)
*/

    // shuffle
    val rdd1 = wordsRDD.map((_, 1)) // 3
    // 再分区  指定分区器 , 数据会按照传入的分区规则分区 [KVRDD]  默认字段 K.hashCode
    val rdd = rdd1.partitionBy(new HashPartitioner(3))
    // 默认会根据HashPartitoner分区器 分区  shuffle
    val rdd2: RDD[(String, Int)] = rdd.reduceByKey( new  MyPartitioner(2) , _ + _) //  3
    rdd.groupByKey(new MyPartitioner(3))
    rdd.groupBy(_._1 , new MyPartitioner(3))
    // 重新分区
    rdd.partitionBy(new MyPartitioner(3))
    // 再去调用计算逻辑


    sc.stop
  }

}
